package com.sunsys.hdfs.compression.gzip;

import java.io.FileInputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionOutputStream;

public class CompressFile {
	
	/**
	 * local source file is read, data will be compressed and will be saved in hdfs location
	 */
	

	public static void main(String []args) {
	
		Configuration configuration = null;
		FileSystem fileSystem = null;
		CompressionCodec compressionCodec = null;
		CompressionCodecFactory compressionCodecFactory = null;
		
		String hdfsUri = "hdfs://localhost:9000/";
		String localSourceFileName = "/home/ubuntu/Desktop/Oct-2012.pdf";
		String hdfsCompressionFileName = "/usr/hdfs/compress/cmpfile1.gz";
		
		FileInputStream localFileInputStream = null;
		CompressionOutputStream compressionOutputStream = null;
		FSDataOutputStream fsDataOutputStream = null;
		try {
			configuration = new Configuration();
			fileSystem = FileSystem.get(new URI(hdfsUri), configuration);
			compressionCodecFactory = new CompressionCodecFactory(configuration);
			
			Path compressionFilePath = new Path(hdfsCompressionFileName);
			fsDataOutputStream = fileSystem.create(compressionFilePath);
			
			
			/* get CompressionCodec instance associated with the file extension */
			compressionCodec = compressionCodecFactory.getCodec(compressionFilePath);
			compressionOutputStream = compressionCodec.createOutputStream(fsDataOutputStream);
			
			localFileInputStream = new FileInputStream(localSourceFileName);
			IOUtils.copyBytes(localFileInputStream, compressionOutputStream, configuration, false);
			
			compressionOutputStream.finish();
		}catch(URISyntaxException e1) {
			e1.printStackTrace();
		}catch(IOException e2) {
			e2.printStackTrace();
		}finally {
			try {
				if( compressionOutputStream != null) {
					compressionOutputStream.close();
				}
				if( localFileInputStream != null) {
					localFileInputStream.close();
				}
				if( fsDataOutputStream == null ) {
					fsDataOutputStream.close();
				}
			}catch(IOException e1){
				e1.printStackTrace();
			}
		}
		
		System.out.println("Thank you");
	}
}
